package com.icarbonx.dap.notification.job

import com.google.gson.Gson
import com.icarbonx.dap.notification.msgservice.rabbitmq.{ConnUtil, MQObject}
import com.icarbonx.dap.notification.util.{DateUtil, KafkaUtil, NoticeMsg, SparkUtil}
import com.icarbonx.dap.notification.vo._
import kafka.serializer.StringDecoder
import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkContext
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Duration, Seconds, StreamingContext}

import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer

/**
 * Created by Lenovo on 2018/8/12.
 * 夜间和空腹  5天内
 * [{
 * "personId":10000001,
 * "value":4.1,
 * "timestamp":1533777079556,
 * "createdTime":1533777079556
 * }]
 */
object BloodSugarFiveDayStream {
  def main(args: Array[String]) {
    Logger.getLogger("org.apache.kafka").setLevel(Level.ERROR)
    Logger.getLogger("org.apache.zookeeper").setLevel(Level.ERROR)
    Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)

    /**
     * brokers = "10.47.250.4:9092,10.47.250.7:9092,10.47.250.6:9092"
     * topicList = "abel_test"
     * mq:10.104.103.7 5672 guest icx@betaOn16 /
     * redis:10.104.103.7 6379 passxx#$#dslllfjshwerw123@ 3000 abel_test
     * brokers = "10.100.240.3:9092,10.100.240.240:9092,10.100.240.2:9092"
     * topicList = "bsmp.blood_glucose"
     * mq:10.0.7.101 5672 guest guest@icx  /
     * redis:172.18.166.122 6379 dev
     */
    println(args.mkString("\t"))
    val Array(brokers, topicList, mqHost, mqPort, mqUsername, mqPassword, mqVirtualHost,
    exchangeName, mqRouteKey, checkpointDir, duration, rddDuration, appName
    ) = args

    //    val checkpointPath = "/data/streaming/checkpoint/bloodsugar_fiveday_stream/"
    val conf = SparkUtil.getSparkConf(appName, Array(classOf[BloodSugarRecord]))

    val sc = SparkContext.getOrCreate(conf)

    val updateFunc = (values: Seq[ArrayBuffer[BloodSugarRecord]],
                      state: Option[ArrayBuffer[BloodSugarRecord]]) => {
      //保存状态
      val todayTimeStamp = DateUtil.getTodayDateTime("yyyyMMddHHmmss") - 12 * 60 * 60 * 1000
      val currentCount = ArrayBuffer[BloodSugarRecord]()
      values.foreach { list => currentCount ++= list }

      val previousCount = state.getOrElse(ArrayBuffer[BloodSugarRecord]())
      Some(currentCount ++ previousCount.filter { record =>
        record.timestamp > todayTimeStamp
      })
    }

    val updateDayCountFunc =
      (values: Seq[mutable.HashMap[String, Int]],
       state: Option[mutable.HashMap[String, Int]]) => {
        //保存状态
        val fiveDaysAgoTimeStamp = DateUtil.getTodayDateTime("yyyyMMddHHmmss") - 5 * 24 * 60 * 60 * 1000
        val currentState = mutable.HashMap[String, Int]()
        val previousState = state.getOrElse(mutable.HashMap[String, Int]())
        values.foreach { case map =>
          map.foreach { case (timeState, flag) =>
            currentState += ((timeState, flag))
          }
        }
        val tempOldState = previousState.filter {
          case (timeState, flag) =>
            val dayTime = timeState.split(",")(0).toLong
            !currentState.contains(timeState) && dayTime > fiveDaysAgoTimeStamp
        }
        Some(currentState ++= tempOldState)
      }

    def creatingFunc(): StreamingContext = {

      val ssc = new StreamingContext(sc, Seconds(duration.toInt))
      ssc.checkpoint(checkpointDir)
      val kafkaParams = KafkaUtil.getKafkaParam(brokers, appName)

      val mqConn = new MQObject(mqHost, mqPort.toInt, mqUsername, mqPassword, mqVirtualHost)

      val topics = topicList.split(",").toSet
      val bloodSugarMessages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc,kafkaParams, topics)

      val newBatchAndOld: DStream[(String, ArrayBuffer[BloodSugarRecord])] =
        bloodSugarMessages
          .filter(_._2.nonEmpty)
          .flatMap { msg =>
            try {
              println("___________ msg :" + msg._2)
              val gson = new Gson()
              val voList =
                gson.fromJson(msg._2, classOf[Array[BloodSugarRecord]])
              voList.map { vo =>
                (vo.personId.toString, ArrayBuffer[BloodSugarRecord](vo))
              }.toList
            } catch {
              case e: Exception =>
                println("____________" + e.getMessage)
                List(("", ArrayBuffer[BloodSugarRecord]()))
            }
          }
          .filter(_._1.nonEmpty)
          .updateStateByKey(updateFunc)

      newBatchAndOld.checkpoint(Duration(rddDuration.toInt * 1000))

      val sortRdd = newBatchAndOld.filter(_._2.nonEmpty)
        .map { case (pid, list) => (pid, list.sortBy(_.timestamp)) }


      val sugarStateByDay = sortRdd
        .map {
          case (pid, list) =>
            println(s"_______ pid:$pid   list " + list.toString())
            val result = mutable.HashMap[String, Int]()
            //夜间低糖
            //            val nightList = list.filter { vo => vo.timestamp >= DateUtil.getTodayATime(0, 0) && vo.timestamp <= DateUtil.getTodayATime(6, 0) }
            val nightList = list.filter { vo =>
              val times = DateUtil.getDateTime(vo.timestamp)
              (times(3) >= 0 && times(3) < 6) || (times(3) == 6 && times(4) == 0)
            }
            if (nightList.size >= 2) {
              var preVo = nightList.head
              for (i <- 1 to nightList.size - 1) {
                val nowVo = nightList(i)
                if (nowVo.timestamp - preVo.timestamp < 15 * 60 * 1000) {
                  if (preVo.value < 2.8 && nowVo.value < 2.8) {
                    result += ((DateUtil.getDateTimestamp(DateUtil.getTimeStampFormat(preVo.timestamp)) + ",1", 1))
                  }
                }
                preVo = nowVo
              }
            }
            //空腹高糖
            //            val morningList = list.filter { vo => vo.timestamp >= DateUtil.getTodayATime(5, 30) && vo.timestamp <= DateUtil.getTodayATime(6, 30) }
            val morningList = list.filter { vo =>
              val times = DateUtil.getDateTime(vo.timestamp)
              (times(3) == 5 && (times(4) >= 30 || times(4) <= 59)) || (times(3) == 6 && (times(4) >= 0 || times(4) <= 30))
            }
            if (morningList.size >= 2) {
              var preVo = morningList.head
              for (i <- 1 to morningList.size - 1) {
                val nowVo = morningList(i)
                if (nowVo.timestamp - preVo.timestamp < 15 * 60 * 1000) {
                  if (preVo.value > 7.8 && nowVo.value > 7.8) {
                    result += ((DateUtil.getDateTimestamp(DateUtil.getTimeStampFormat(preVo.timestamp)) + ",2", 1))
                  }
                }
                preVo = nowVo
              }
            }
            //午餐前
            //            val lunchList = list.filter { vo => vo.timestamp >= DateUtil.getTodayATime(11, 30) && vo.timestamp <= DateUtil.getTodayATime(12, 30) }
            val lunchList = list.filter { vo =>
              val times = DateUtil.getDateTime(vo.timestamp)
              (times(3) == 11 && (times(4) >= 30 || times(4) <= 59)) || (times(3) == 12 && (times(4) >= 0 || times(4) <= 30))
            }

            if (lunchList.size > 1) {
              var preVo = lunchList.head
              for (i <- 1 to lunchList.size - 1) {
                val nowVo = lunchList(i)
                if (preVo.value < 3 && nowVo.value < 3
                  && nowVo.timestamp - preVo.timestamp < 15 * 60 * 60 * 1000) {
                  result += ((DateUtil.getDateTimestamp(DateUtil.getTimeStampFormat(preVo.timestamp)) + ",3", 1))
                }
                preVo = nowVo
              }
            }
            //晚餐前
            //            val dinnerList = list.filter { vo => vo.timestamp >= DateUtil.getTodayATime(17, 0) && vo.timestamp <= DateUtil.getTodayATime(19, 0) }
            val dinnerList = list.filter { vo =>
              val times = DateUtil.getDateTime(vo.timestamp)
              (times(3) >= 17 && times(3) < 19) || (times(3) == 19 && times(4) == 0)
            }

            if (dinnerList.size > 1) {
              var preVo = dinnerList.head
              for (i <- 1 to dinnerList.size - 1) {
                val nowVo = dinnerList(i)
                if (preVo.value < 3 && nowVo.value < 3
                  && nowVo.timestamp - preVo.timestamp < 15 * 60 * 60 * 1000) {
                  result += ((DateUtil.getDateTimestamp(DateUtil.getTimeStampFormat(preVo.timestamp)) + ",4", 1))
                }
                preVo = nowVo
              }
            }

            (pid, result)
        }
        .filter(_._2.nonEmpty)
        .updateStateByKey(updateDayCountFunc)

      sugarStateByDay.checkpoint(Duration(rddDuration.toInt * 1000))

      sugarStateByDay
        .filter(_._2.nonEmpty)
        .foreachRDD(mess =>
          if (!mess.isEmpty()) {
            mess.foreachPartition { lines =>
              //              println("_______ lines:" + lines.toString())
              val mqMsg = ArrayBuffer[String]()
              val gson = new Gson()
              lines.foreach {
                case (pid, list) =>
                  println(s"_______ pid:$pid   list " + list.toString())
                  //@todo 16 连续两个点血糖<3.0mmol/L,5天内出现3次&0:00~6:00
                  if (list.filter(_._1.endsWith("1")).values.sum >= 3) {
                    val msg = new WarningMsg(
                      "glycemia",
                      pid.toLong,
                      NoticeMsg.low_glycemia_night_freq._1,
                      NoticeMsg.low_glycemia_night_freq._2,
                      NoticeMsg.low_glycemia_night_freq._3,
                      list.filter(_._1.endsWith("1")).keys.max.dropRight(2),
                      NoticeMsg.low_glycemia_night_freq._5,
                      NoticeMsg.low_glycemia_night_freq._4,
                      NoticeMsg.low_glycemia_night_freq._6
                    )
                    mqMsg += gson.toJson(msg)
                  }
                  //@todo 22 6:00&前后两个点血糖>7.8mmol/L，5内出现3次
                  if (list.filter(_._1.endsWith("2")).values.sum >= 3) {
                    val msg = new WarningMsg(
                      "glycemia",
                      pid.toLong,
                      NoticeMsg.high_glycemia_fast_freq._1,
                      NoticeMsg.high_glycemia_fast_freq._2,
                      NoticeMsg.high_glycemia_fast_freq._3,
                      list.filter(_._1.endsWith("2")).keys.max.dropRight(2),
                      NoticeMsg.high_glycemia_fast_freq._5,
                      NoticeMsg.high_glycemia_fast_freq._4,
                      NoticeMsg.high_glycemia_fast_freq._6
                    )
                    mqMsg += gson.toJson(msg)
                  }
                  //@todo 19.多发午餐前低糖警报:连续两个点血糖<3.0mmol/L,5天内出现3次&既定午餐时间前1小时内:low_glycemia_lunch_freq:每天午餐前:触发及时发送，每小时提示不超过1条
                  if (list.filter(_._1.endsWith("3")).values.sum >= 3) {
                    val msg = new WarningMsg(
                      "glycemia",
                      pid.toLong,
                      NoticeMsg.low_glycemia_lunch_freq._1,
                      NoticeMsg.low_glycemia_lunch_freq._2,
                      NoticeMsg.low_glycemia_lunch_freq._3,
                      list.filter(_._1.endsWith("3")).keys.max.dropRight(2),
                      NoticeMsg.low_glycemia_lunch_freq._5,
                      NoticeMsg.low_glycemia_lunch_freq._4,
                      NoticeMsg.low_glycemia_lunch_freq._6
                    )
                    mqMsg += gson.toJson(msg)
                  }

                  //@todo 20.多发晚餐前低糖警报:连续两个点血糖<3.0mmol/L,5天内出现3次&既定晚餐时间前2小时内:low_glycemia_supper_freq:每天晚餐前:触发及时发送，每小时提示不超过1条
                  if (list.filter(_._1.endsWith("4")).values.sum >= 3) {
                    val msg = new WarningMsg(
                      "glycemia",
                      pid.toLong,
                      NoticeMsg.low_glycemia_supper_freq._1,
                      NoticeMsg.low_glycemia_supper_freq._2,
                      NoticeMsg.low_glycemia_supper_freq._3,
                      list.filter(_._1.endsWith("4")).keys.max.dropRight(2),
                      NoticeMsg.low_glycemia_supper_freq._5,
                      NoticeMsg.low_glycemia_supper_freq._4,
                      NoticeMsg.low_glycemia_supper_freq._6
                    )
                    mqMsg += gson.toJson(msg)
                  }
              }

              if (mqMsg.nonEmpty) {
                val mqChannel = mqConn.connect().createChannel()
                mqMsg.foreach { msg =>
                  ConnUtil.sendMsg(mqChannel, exchangeName, mqRouteKey, msg)
                }
                mqChannel.close()
              }
            }

          })

      ssc
    }


    val ssc = StreamingContext.getOrCreate(checkpointDir, creatingFunc)

    ssc.start()
    ssc.awaitTermination()
  }

}